package com.dxy.chatserver.builder;

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.dxy.chatserver.config.RocketConfig;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.function.Function;

/**
 * 生成配置类
 *
 * @author: none
 * @date: 2021/12/8
 **/
@Slf4j
public class RocketBuilder {

    /**
     * Mapper
     */
    public static final ObjectMapper MAPPER = new ObjectMapper();

    /**
     * 初始化配置
     *
     * @param accessKey          access key
     * @param secretKey          secret key
     * @param consumerThreadNums 消费端处理线程数
     * @param nameServiceAddress name service address
     * @param groupId            group id
     * @param messageModel       消费模式 @link PropertyValueConst
     * @return 配置信息
     */
    public static Properties properties(String accessKey, String secretKey, String nameServiceAddress, Integer consumerThreadNums, String groupId, String messageModel) {
        Properties properties = new Properties();
        // 您在控制台创建的Group ID。
        if (Objects.nonNull(groupId) && !groupId.isEmpty()) {
            properties.setProperty(PropertyKeyConst.GROUP_ID, groupId);
        }
        // AccessKey ID阿里云身份验证，在阿里云RAM控制台创建。
        properties.setProperty(PropertyKeyConst.AccessKey, accessKey);
        // AccessKey Secret阿里云身份验证，在阿里云服RAM控制台创建。
        properties.setProperty(PropertyKeyConst.SecretKey, secretKey);
        // 设置TCP接入域名，进入控制台的实例详情页面的TCP协议客户端接入点区域查看。
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, nameServiceAddress);
        // 广播订阅方式(默认)。
        if (Objects.nonNull(messageModel) && !messageModel.isEmpty()) {
            properties.put(PropertyKeyConst.MessageModel, messageModel);
        }
        // 消息发送超时时间，如果服务端在配置的对应时间内未ACK，则发送客户端认为该消息发送失败
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "60000");
        properties.setProperty(PropertyKeyConst.ConsumeTimeout, "10");
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, consumerThreadNums.toString());
        properties.setProperty(PropertyKeyConst.BatchConsumeMaxAwaitDurationInSeconds, "0");
        return properties;
    }

    /**
     * 构建生产者
     *
     * @param config          配置信息
     * @param group           组
     * @param msgMessageModel 模式
     * @return
     */
    public static Producer httpProducer(RocketConfig config, String group, String msgMessageModel) {
        return ONSFactory.createProducer(RocketBuilder.properties(config.getAccessKey(), config.getSecretKey(), config.getNameServiceAddress(), config.getConsumeThreadNums(), group, msgMessageModel));
    }

    /**
     * 构建生产者
     *
     * @param config 配置信息
     * @return
     */
    public static Producer tcpProducer(RocketConfig config) {
        return ONSFactory.createProducer(RocketBuilder.properties(config.getAccessKey(), config.getSecretKey(), config.getNameServiceAddress(), config.getConsumeThreadNums(), null, null));
    }

    /**
     * 消费者
     * - 单条消息
     *
     * @param config          配置文件
     * @param group           组
     * @param topic           topic
     * @param msgMessageModel 模式
     * @param function        处理方法
     * @param clazz           类型
     * @param <T>             类型
     * @return
     */
    public static <T> ConsumerBean singleMsg(RocketConfig config, String group, String topic, String msgMessageModel, Function<T, Boolean> function, Class<T> clazz) {
        ConsumerBean consumerBean = new ConsumerBean();
        consumerBean.setProperties(RocketBuilder.properties(config.getAccessKey(), config.getSecretKey(), config.getNameServiceAddress(), config.getConsumeThreadNums(), group, msgMessageModel));
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();
        Subscription subscription = new Subscription();
        subscription.setTopic(topic);
        subscriptionTable.put(subscription, (message, consumeContext) -> {
            try {
                if (function.apply(MAPPER.readValue(message.getBody(), clazz))) {
                    return Action.CommitMessage;
                } else {
                    return Action.ReconsumeLater;
                }
            } catch (Exception e) {
                log.error("消息消费异常! --> ", e);
                return Action.ReconsumeLater;
            }
        });
        // 订阅多个topic如上面设置
        consumerBean.setSubscriptionTable(subscriptionTable);
        return consumerBean;
    }

    /**
     * object to json
     *
     * @param object 实体
     * @return
     */
    public static String writeValueAsString(Object object) {
        try {
            return MAPPER.writeValueAsString(object);
        } catch (Exception e) {
            log.error("object to json error! --> object = {}", object, e);
            throw new RuntimeException("object to json error!");
        }
    }

}
